from urllib.parse import quote_plus

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.testing.plugin.plugin_base import engines


def redis_test():
    import redis



    redis_conn = redis.StrictRedis(host='r-bp1u7oiaxwfpof7014.redis.rds.aliyuncs.com', port=6379,
                                                password='GHGOAPSqVaMWZJf6JCT', db=0)




    redis_conn.set('dy_message_qrcode:{}'.format("doy_test_1"), "556453")
    print("写入成功")
    redis_conn.expire('dy_message_qrcode:{}'.format("doy_test_1"), 60)
    print('"se')
    return "成功"


redis_test()

def mysql_test():


    user = 'warehouse'
    # password = quote_plus('QlciCnfTB9NtLSC')
    password = 'QlciCnfTB9NtLSC'
    # passwd ='merchantsasd123!@%&'
    host = 'rm-bp17709mm0o30s13r.mysql.rds.aliyuncs.com'
    # port1 ='3306'
    dbname2 = 'data-warehouse'
    engine2 = create_engine(f"mysql+pymysql://{user}:{password}@{host}:3306/{dbname2}?charset=utf8mb4")

    import pandas as pd

    # 定义一些示例数据
    data = {
        '姓名': ['张三', '李四', '王五', '赵六'],
        '年龄': [23, 25, 22, 24],
        '城市': ['北京', '上海', '广州', '深圳'],
        '分数': [88, 92, 95, 85]
    }

    # 创建一个 DataFrame

    df = pd.DataFrame(data)


    pd.io.sql.to_sql(df, 'mysql_dy_test', engine2, schema='warehouse',
                             if_exists='append',
                             index=False)

# mysql_test()

def mysql_test_read():


    user = 'warehouse'
    # password = quote_plus('QlciCnfTB9NtLSC')
    password = 'QlciCnfTB9NtLSC'
    # passwd ='merchantsasd123!@%&'
    host = 'rm-bp17709mm0o30s13r.mysql.rds.aliyuncs.com'
    # port1 ='3306'
    dbname2 = 'data-warehouse'
    engine2 = create_engine(f"mysql+pymysql://{user}:{password}@{host}:3306/{dbname2}?charset=utf8mb4")

    data = pd.read_sql("select * from dy_user_info",engine2)
    print(data)
# mysql_test_read()